package com.pz.chat.netty_ws.handlers;

import com.alibaba.fastjson2.JSON;
import com.pz.chat.constant.MessageCodeEnum;
import com.pz.chat.constant.RabbitMqConstant;
import com.pz.chat.entity.ChatMessageEntity;
import com.pz.chat.constant.MessageTypeEnum;
import com.pz.chat.config.netty.NettyConfig;
import com.pz.chat.service.ChatMessageService;
import com.pz.chat.utils.RedisUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.stream.Collectors;

/**
 *  1、在建立ws时需要发送一个注册用户userId的消息，type参数必须携带，且为 0
       {
          "fromId": '001',
          "toId": '',
          "msgContent": '',
          "sendTime": null,
          "type": 0
       }
    2、选择聊天对象时必发送的消息格式, type参数必须携带，且为 2，注册聊天对象
     {
         "fromId": '001',
         "toId": '002',
         "msgContent": '', // 可能会为文件格式（图片、视频等）
         "sendTime": null,
         "type": 2
     }
 */

@Slf4j
@Component
// 保证处理器，在整个生命周期以单例形式存在，方便统计客户端在线数量
@ChannelHandler.Sharable
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

    @Autowired
    private RedisUtil redisUtil;
    private final String WEB_SOCKET_URL = "ws://"+ this.discoveryIp+this.wsPath;

    private WebSocketServerHandshaker webSocketServerHandshaker;

    @Value("${netty.websocket.server.port}")
    private String webSocketServerPort;
    @Value("${netty.websocket.server.discoveryIP}")
    private String discoveryIp;
    @Value("${netty.websocket.server.path}")
    private String wsPath;

    @Autowired
    private NettyConfig nettyConfig;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 创建连接时执行
        log.info("client channel active, id={}", ctx.channel().id().toString());
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete) {
            WebSocketServerProtocolHandler.HandshakeComplete handshake = (WebSocketServerProtocolHandler.HandshakeComplete) evt;
            log.info("client channel connected, id={}, url={}", ctx.channel().id().toString(), handshake.requestUri());
        }
    }

    /**
     *  读取客户端数据
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest) {
            // 处理客户端http握手请求
            handlerHttpRequest(ctx, (FullHttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {
            // 处理websocket连接业务
            handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        // 关闭连接时执行
        String channelId =  ctx.channel().id().toString();
        String userId = nettyConfig.getChannelMap().get(channelId);
        log.info("client channel disconnected, id={}; user disconnected, userId-{}", channelId, userId);
        nettyConfig.fresh(ctx);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        // 服务端接收客户端发送过来的数据结束之后调用
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 非正常断开时调用
        log.error("client channel execute exception, id={}, reason-{}", ctx.channel().id().toString(), cause.getMessage());
        nettyConfig.fresh(ctx);
        ctx.close();
    }

    /**
     * 处理websocket连接业务
     */
    private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        // 判断是否是关闭websocket的指令
        if (frame instanceof CloseWebSocketFrame) {
            webSocketServerHandshaker.close(ctx.channel(), ((CloseWebSocketFrame) frame).retain());
            return;
        }
        // 判断是否是ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        /* 处理文本型数据 */
        if (frame instanceof TextWebSocketFrame) {
            String text = ((TextWebSocketFrame) frame).text();
            if ("ping".equals(text)) {
                ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
                return;
            }
            ChatMessageEntity chatMessage;
            try {
                chatMessage = JSON.to(ChatMessageEntity.class, text);
            } catch (Exception e) {
                log.error("Illegal message format -- {}", text);
                ChatMessageEntity resMsg = new ChatMessageEntity("0", "0",
                        MessageCodeEnum.ILLEGAL_MSG_FROMAT.getMsg(),
                        MessageTypeEnum.SERVER_INFO.getType(), MessageCodeEnum.ILLEGAL_MSG_FROMAT.getCode(),
                        1
                );
                sendTextMsg(ctx.channel(), resMsg);
                return;
            }

            if (chatMessage.getFromId() != null && chatMessage.getType() != null) {
                switch (chatMessage.getType()) {
                    case 0:
                        /* 将客户端用户的 userId与channelId建立map映射，同步到Redis, 如果是服务的client就以 前缀+ip:port 为key **/
                        if(redisUtil.hasKey(chatMessage.getFromId())) {
                            nettyConfig.fresh(ctx);
                        }
                        nettyConfig.getChannelGroup().add(ctx.channel());
                        nettyConfig.getClientMap().put(chatMessage.getFromId(), ctx.channel().id());
                        nettyConfig.getChannelMap().put(ctx.channel().id().toString(), chatMessage.getFromId());
                        log.info("local register channel-{}, channelId-{}", chatMessage.getFromId(), ctx.channel().id());
                        redisUtil.set(nettyConfig.getClientServerBindKey(chatMessage.getFromId()), discoveryIp+"-"+webSocketServerPort+"-"+ctx.channel().id().toString());
                        break;
                    case 1: /* 好友申请 **/
                    case 2: /*注册聊天对象ID(每次切换聊天对象必触发的方法)*/
                        nettyConfig.getChannelTargetMap().put(chatMessage.getFromId(), chatMessage.getToId());
                        break;
                    case 4: /* 客户端确认收到消息(这些消息已经存在在数据库中了，只是) **/
                        chatMessage.setStatus(1);
                        break;
                    case 5: /* 客户端消息私发 **/
                        Channel targetChannel = searchTargetCh(chatMessage.getToId());
                        if (targetChannel == null) {
                            chatMessage.setStatus(0);
                        } else {
                            sendTextMsg(targetChannel, chatMessage);
                        }
                        break;
                    default:
                        log.error("message type code illegal!!! failed type code-{}", chatMessage.getType());
                        break;
                }
                // 发送消息给MQ，削峰，避免直接存储造成数据库奔溃(只存储非注册消息)
                if (chatMessage.getType().intValue() != MessageTypeEnum.CLIENT_INIT_USERID.getType()) {
                    rabbitTemplate.convertAndSend(RabbitMqConstant.chatMsgExchange, RabbitMqConstant.chatMsgRoutingKey, chatMessage);
                }
            } else {
                String errMessage = "";
                if (chatMessage.getType() == null) {
                    errMessage = "消息 type 参数必须携带";
                }
                if(chatMessage.getFromId() == null) {
                    errMessage = "消息 fromId 参数必须携带";
                }
                ChatMessageEntity resMsg = new ChatMessageEntity("0", chatMessage.getFromId(), errMessage, MessageTypeEnum.SERVER_INFO.getType(),
                        MessageCodeEnum.ILLEGAL_MSG_FROMAT.getCode(), 1);
                sendTextMsg(ctx.channel(), resMsg);
            }
            return;
        }

        /* 处理二进制数据 */
        if(frame instanceof BinaryWebSocketFrame) {
            BinaryWebSocketFrame binaryFrame = (BinaryWebSocketFrame)frame;
            ByteBuf content = binaryFrame.content();
            log.info("receive binary msg, the len of msg is {}", content.capacity());
            String userId = nettyConfig.getChannelMap().get(ctx.channel().id().toString());
            if(binaryFrame.isFinalFragment()){
                Channel targetChannel = searchTargetCh(nettyConfig.getChannelTargetMap().get(userId));
                sendBinaryMsg(targetChannel, content);
            } else {
                ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
                byteBuf.writeBytes(content);
                nettyConfig.getByteBufMap().put(userId, byteBuf);
            }
            return;
        }

        /* 处理没有发完的包(此处处理二进制数据) */
        if(frame instanceof ContinuationWebSocketFrame){
            ContinuationWebSocketFrame continuationFrame = (ContinuationWebSocketFrame) frame;
            ByteBuf content = continuationFrame.content();
            String userId = nettyConfig.getChannelMap().get(ctx.channel().id().toString());
            ByteBuf byteBufBuild = nettyConfig.getByteBufMap().get(userId);

            if(content != null) {
                byteBufBuild.writeBytes(content);
            }

            if (continuationFrame.isFinalFragment()) {
                // 最后一个片段已到达，可以将消息发送到下一个处理阶段
                if(byteBufBuild != null){
                    Channel targetChannel = searchTargetCh(nettyConfig.getChannelTargetMap().get(userId));
                    sendBinaryMsg(targetChannel, byteBufBuild);
                    nettyConfig.getByteBufMap().remove(userId);
                }
            }
            return;
        }

        throw new RuntimeException("不支持消息类型：" + frame.getClass().getName());
    }

    /**
     * 处理客户端http握手请求
     */
    private void handlerHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {
        log.info("handlerHttpRequest>>>>class={}", request.getClass().getName());
        // 判断是否采用WebSocket协议
        if (!request.decoderResult().isSuccess() || !("websocket".equals(request.headers().get("Upgrade")))) {
            sendHttpResponse(ctx, request, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }

        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(WEB_SOCKET_URL, null, false, 65536*5);
        webSocketServerHandshaker = wsFactory.newHandshaker(request);
        if (webSocketServerHandshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            webSocketServerHandshaker.handshake(ctx.channel(), request);
        }
    }

    /**
     * Http 响应
     */
    private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest request, DefaultFullHttpResponse response) {
        if (response.status().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(response.status().toString(), StandardCharsets.UTF_8);
            response.content().writeBytes(buf);
            buf.release();
        }
        // 服务端向客户端发送数据
        ChannelFuture f = ctx.channel().writeAndFlush(response);
        if (response.status().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    /**
     * 获取聊天对象的 channel
     * @param toId 聊天对象ID
     */
    private Channel searchTargetCh(String toId) {
        // 判断客户端是否在线, 不在线就通知客户端
        Channel targetChannel = nettyConfig.getChannel(toId);
        String otherServerCh = (String) redisUtil.get(nettyConfig.getClientServerBindKey(toId));
        if (targetChannel != null) {
            // 本地channel
            return targetChannel;
        }else if (otherServerCh != null) {
            // 目标聊天对象的 channel 在其他的服务器上，向这个服务器转发消息
            String[] split = otherServerCh.split("-");
            targetChannel = nettyConfig.getChannel(NettyConfig.serverClientKey(split[0], Integer.parseInt(split[1])));
            if (targetChannel != null) {
                return targetChannel;
            } else {
                log.error("redis 与 本地的channel信息不一致 {}", Arrays.toString(split));
                return null;
            }
        } else {
            return null;
        }
    }

    /**
     * 发送文本型消息
     * @param channel 获取聊天对象的 channel
     * @param chatMessage 消息体
     */
    private void sendTextMsg(Channel channel, ChatMessageEntity chatMessage) {
        channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(chatMessage)));
    }

    private void sendBinaryMsg(Channel channel, ByteBuf byteBuf){
        if(channel != null) {
            channel.writeAndFlush(new BinaryWebSocketFrame(byteBuf));
        } else {
            //TODO 聊天对象不在线的二进制数据处理
            log.debug("target is offline");
        }
        //TODO 处理二进制数据,持久存储 */
        InputStream byteBufInputStream = new ByteBufInputStream(byteBuf);
    }
}
